/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package remoting.netty4.client;

import common.constants.CommonConstants;
import common.model.Codec2;
import common.model.MagicUrl;
import common.model.Request;
import common.model.Response;
import common.util.StringUtils;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.channel.ChannelHandler;
import remoting.netty4.NettyChannel;
import serialization.Serializer;

/**
 * NettyClientHandler
 */
@io.netty.channel.ChannelHandler.Sharable
public class NettyClientHandler extends ChannelDuplexHandler {

  private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);

  private final MagicUrl url;

  private final ChannelHandler handler;

  private Codec2 codec;

  public NettyClientHandler(MagicUrl url, ChannelHandler handler, Codec2 codec) {
    if (url == null) {
      throw new IllegalArgumentException("url == null");
    }
    if (handler == null) {
      throw new IllegalArgumentException("handler == null");
    }
    this.url = url;
    this.handler = handler;
    this.codec = codec;
  }

  @Override
  public void channelActive(ChannelHandlerContext ctx) throws Exception {
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler, codec);
    try {
      handler.connected(channel);
    } finally {
      NettyChannel.removeChannelIfDisconnected(ctx.channel());
    }
  }

  @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler, codec);
    try {
      handler.disconnected(channel);
    } finally {
      NettyChannel.removeChannelIfDisconnected(ctx.channel());
    }
  }

  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (logger.isDebugEnabled()) {
      logger.debug("client read {}", msg);
    }
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler, codec);
    try {
      handler.received(channel, msg);
    } finally {
      NettyChannel.removeChannelIfDisconnected(ctx.channel());
    }
  }

  @Override
  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
      throws Exception {
    if (logger.isDebugEnabled()) {
      logger.debug("client send {}", msg);
    }
    super.write(ctx, msg, promise);
    final NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler, codec);
    final boolean isRequest = msg instanceof Request;

    promise.addListener(future -> {
      try {
        if (future.isSuccess()) {
          handler.sent(channel, msg);
          return;
        }

        Throwable t = future.cause();
        if (t != null && isRequest) {
          Request request = (Request) msg;
          Response response = buildErrorResponse(request, t);
          handler.received(channel, response);
        }
      } finally {
        NettyChannel.removeChannelIfDisconnected(ctx.channel());
      }
    });
  }

  @Override
  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    // send heartbeat when read idle.
    if (evt instanceof IdleStateEvent) {
      try {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler, codec);
        if (logger.isDebugEnabled()) {
          logger.debug("IdleStateEvent triggered, send heartbeat to channel " + channel);
        }
        Request req = new Request();
        req.setVersion(CommonConstants.PROTOCOL_VERSION);
        req.setTwoWay(true);
        req.setEvent(Request.HEARTBEAT_EVENT);
        channel.sendMessage(req);
      } finally {
        NettyChannel.removeChannelIfDisconnected(ctx.channel());
      }
    } else {
      super.userEventTriggered(ctx, evt);
    }
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
      throws Exception {
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler, codec);
    try {
      handler.caught(channel, cause);
    } finally {
      NettyChannel.removeChannelIfDisconnected(ctx.channel());
    }
  }

  /**
   * build a bad request's response
   *
   * @param request the request
   * @param t       the throwable. In most cases, serialization fails.
   * @return the response
   */
  private static Response buildErrorResponse(Request request, Throwable t) {
    Response response = new Response(request.getId(), request.getVersion());
    response.setStatus(Response.BAD_REQUEST);
    response.setErrorMessage(StringUtils.toString(t));
    return response;
  }
}
